package com.autonavi.yunda.yunji.core.engine.handler;

import java.util.List;

import javax.annotation.PostConstruct;

import com.aliyuncs.DefaultAcsClient;
import com.aliyuncs.IAcsClient;
import com.aliyuncs.dataworks.model.v20200918.CreateDagComplementRequest;
import com.aliyuncs.dataworks.model.v20200918.CreateDagComplementResponse;
import com.aliyuncs.dataworks.model.v20200918.GetDagRequest;
import com.aliyuncs.dataworks.model.v20200918.GetDagResponse;
import com.aliyuncs.dataworks.model.v20200918.ListInstancesRequest;
import com.aliyuncs.dataworks.model.v20200918.ListInstancesResponse;
import com.aliyuncs.dataworks.model.v20200918.RestartInstanceRequest;
import com.aliyuncs.dataworks.model.v20200918.RestartInstanceResponse;
import com.aliyuncs.dataworks.model.v20200918.ResumeInstanceRequest;
import com.aliyuncs.dataworks.model.v20200918.ResumeInstanceResponse;
import com.aliyuncs.exceptions.ClientException;
import com.aliyuncs.profile.DefaultProfile;
import com.aliyuncs.profile.IClientProfile;
import com.google.common.util.concurrent.RateLimiter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

/**
 * @author liwenjie
 * @date 2024/03/13
 */
@Slf4j
@Component
public class DataOsHandler {

    private IAcsClient dataWorksInnerClient;

    /**
     * DataWorks接口限流1QPS
     */
    private RateLimiter limiter = RateLimiter.create(1);

    private String projectName;

    private String endPoint;

    private String accessId;

    private String accessKey;

    @PostConstruct
    public void init() {
        try {

            IClientProfile dataWorksInnerProfile = DefaultProfile.getProfile("cn-hangzhou", "LTAI5tGzzFv2b8BkhAZMHcbw",
                "AOwp0JqcORsyW02on7jBtQlcPnT8GV");
            DefaultProfile.addEndpoint("cn-hangzhou", "dataworks", "dataworks-share.aliyuncs.com");

            dataWorksInnerClient = new DefaultAcsClient(dataWorksInnerProfile);
        } catch (Exception e) {
            log.error("初始化dataWorks inner client异常", e);
        }
    }

    /**
     * 触发回补数据流程
     * @param flowName 流程名称
     * @param params  各节点参数json
     * @return
     */
    /**
     * 触发回补数据流程
     *
     * @param flowName       流程名称
     * @param startBizDate   开始时间
     * @param endBizDate     结束时间
     * @param rootNodeId     起始节点
     * @param nodeParams     节点参数
     * @param includeNodeIds 节点列表
     * @return
     */
    public List<Long> createDagComplement(String flowName, String startBizDate, String endBizDate, Long rootNodeId,
        String nodeParams, String includeNodeIds) {
        List<Long> dagId = null;
        limiter.acquire();
        try {
            CreateDagComplementRequest request = new CreateDagComplementRequest();
            request.setProjectEnv("PROD");
            request.setName(flowName);
            request.setParallelism(false);
            request.setStartBizDate(startBizDate);
            request.setEndBizDate(endBizDate);
            request.setRootNodeId(rootNodeId);
            request.setNodeParams(nodeParams);
            request.setIncludeNodeIds(includeNodeIds);
            CreateDagComplementResponse response = dataWorksInnerClient.getAcsResponse(request);
            log.info(String.valueOf(response));
            dagId = response.getData();
        } catch (ClientException e) {
            log.error("触发dataWorks手动任务[{}]任务失败， params is [{}]", flowName, nodeParams, e);
        }
        return dagId;
    }

    /**
     * 根据补数据、手动业务流程的流程id查询实例列表
     * @param dagId
     * @param nodeId
     * @param projectId
     * @return
     * @throws ClientException
     */
    public ListInstancesResponse listInstances(Long dagId, Long nodeId, Long projectId) throws ClientException {
        ListInstancesRequest request = new ListInstancesRequest();
        request.setDagId(dagId);
        request.setNodeId(nodeId);
        request.setProjectId(projectId);
        request.setProjectEnv("PROD");
        return dataWorksInnerClient.getAcsResponse(request);
    }

    /**
     * 恢复暂停状态的实例
     * @param instanceId
     * @return
     * @throws ClientException
     */
    public ResumeInstanceResponse resumeInstance(Long instanceId) throws ClientException {
        ResumeInstanceRequest request = new ResumeInstanceRequest();
        request.setInstanceId(instanceId);
        request.setProjectEnv("PROD");
        return dataWorksInnerClient.getAcsResponse(request);
    }

    /**
     * 重新运行实例
     * @param instanceId
     * @return
     * @throws ClientException
     */
    public RestartInstanceResponse restartInstance(Long instanceId) throws ClientException {
        RestartInstanceRequest request = new RestartInstanceRequest();
        request.setInstanceId(instanceId);
        request.setProjectEnv("PROD");
        return dataWorksInnerClient.getAcsResponse(request);
    }

    public String checkDagStatus(Long dagId) {
        try {
            limiter.acquire();
            // 轮询dag状态，直到dag成功或失败
            GetDagRequest getDagRequest = new GetDagRequest();
            getDagRequest.setDagId(dagId);
            getDagRequest.setProjectEnv("PROD");
            GetDagResponse dagResponse = dataWorksInnerClient.getAcsResponse(getDagRequest);
            log.info(dagResponse.toString());
            return dagResponse.getData().getStatus();
        } catch (Exception e) {
            log.error("检查任务流程状态异常，dagId is [{}]", dagId, e);
        }
        return null;
    }

}
